package org.ht.demo.openapi.mqtt;

import java.util.ArrayList;
import java.util.List;

import javax.annotation.PostConstruct;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 鸿途科技
 * @param args
 */


/**
 * mqtt服务
 */
public class MqttServices {

	private String host="tcp://ip:1883"; // 服务地址
	private String userName=""; // 用户名
	private String passWord=""; // 密码
	private int keepAlive=10; // 超时时间
	private int keepAliveInterval=20; // 会话心跳时间
	private String clientId="xx-mqtt-dq-00001"; // 客户端ID
	public static MqttClient client; // 客户端对象





	//主题
	public static List<String> slist = new ArrayList<String>() {{
	    add("/xx/dqv1");    //用电-断路器-安全用电探测器
	    add("/xx/ygv1");    //消防火灾等设备（烟感，手报，声光，门磁，水浸，煤气、天然气气体探测）
		add("/xx/swhv3");    //消防水设备（水位变送器，压力变送器，消火栓）

	}};
	
	public void inint() {
		try {
			// 构建MQTT连接
			boolean cb = connect(true);
			if (cb) {
				System.out.println("mqtt初始化---");
				// 订阅设备  
				for (String st : slist) {
					subScription(st,1);
				}
			}
		} catch (Exception e) {
			System.out.println("mqtt初始化---"+e);
			e.printStackTrace();
		}
	}

	/**
	 * 连接服务器
	 * 
	 * @param cleanSession 是否清除session
	 */
	public boolean connect(boolean cleanSession) {
		try {
			client = new MqttClient(host, clientId, new MemoryPersistence());
			MqttConnectOptions options = new MqttConnectOptions();
			options.setCleanSession(cleanSession);
			options.setUserName(userName);
			options.setPassword(passWord.toCharArray());
			// 设置超时时间
			options.setConnectionTimeout(keepAlive);
			// 设置会话心跳时间
			options.setKeepAliveInterval(keepAliveInterval);
			client.setCallback(new MessageCallback());
			client.connect(options);
			System.out.println("mqtt服务连接成功！！！");
			return true;
		} catch (Exception e) {
			System.out.println("mqtt服务连接失败！！！"+ e);
			return false;
		}
	}

	/**
	 * 断开客户端连接
	 * 
	 */
	public boolean disConnect() {
		try {
			client.disconnect();
			client.close();
			System.out.println("mqtt服务断开成功！！！");
			return true;
		} catch (MqttException e) {
			System.out.println("mqtt服务断开成功！！！"+e);
			return false;
		}
	}

	/**
	 * 主题订阅
	 * 
	 * @param topics 主题名称数组
	 * @param Qos    服务质量数组（与主题名称一一匹配）
	 * @return
	 */
	public boolean subScription(String[] topics, int[] Qos) {
		try {
			client.subscribe(topics, Qos);
			String loginfo = "mqtt主题{";
			for (int i = 0; i < topics.length; i++) {
				loginfo += topics[i];
				if (i != topics.length - 1)
					loginfo += ",";
			}
			loginfo += "}订阅成功！！！";
			System.out.println(loginfo);
			return true;
		} catch (Exception e) {
			String loginfo = "mqtt主题{";
			for (int i = 0; i < topics.length; i++) {
				loginfo += topics[i];
				if (i != topics.length - 2)
					loginfo += ",";
			}
			loginfo += "}订阅失败！！！";
			System.out.println(loginfo);
			return false;
		}
	}

	/**
	 * 主题订阅
	 * 
	 * @param topic 主题名称
	 * @param Qos   服务质量
	 * @return
	 */
	public static boolean subScription(String topic, int Qos) {
		try {
			int[] _Qos = { Qos };
			String[] _topic = { topic };
			client.subscribe(_topic, _Qos);
			System.out.println("mqtt主题{" + topic + "}订阅成功！！！");
			return true;
		} catch (Exception e) {
			System.out.println("mqtt主题{" + topic + "}订阅失败！！！"+ e);
			return false;
		}
	}

	/**
	 * 主题订阅，默认服务质量为0
	 * 
	 * @param topic 主题名称
	 * @return
	 */
	public boolean subScription(String topic) {
		try {
			int[] _Qos = { 0 };
			String[] _topic = { topic };
			client.subscribe(_topic, _Qos);
			System.out.println("mqtt主题{" + topic + "}订阅成功！！！");
			return true;
		} catch (Exception e) {
			System.out.println("mqtt主题{" + topic + "}订阅失败！！！"+ e);
			return false;
		}
	}

	/**
	 * 取消主题订阅
	 * 
	 * @param topics 主题名称
	 * @return
	 */
	public boolean unSubScription(String[] topics) {
		String topic = "";
		try {
			for (int i = 0; i < topics.length; i++) {
				topic = topics[i];
				client.unsubscribe(topic);
				System.out.println("mqtt主题{" + topic + "}取消订阅成功！！！");
			}
			return true;
		} catch (Exception e) {
			System.out.println("mqtt主题{" + topic + "}取消订阅失败！！！"+e);
			return false;
		}
	}

	/**
	 * 取消主题订阅
	 * 
	 * @param topic 主题名称
	 * @return
	 */
	public boolean unSubScription(String topic) {
		try {
			client.unsubscribe(topic);
			System.out.println("mqtt主题{" + topic + "}取消订阅成功！！！");
			return true;
		} catch (Exception e) {
			System.out.println("mqtt主题{" + topic + "}取消订阅失败！！！"+e);
			return false;
		}
	}

	/**
	 * 推送消息
	 * 
	 * @param topicName 主题名称
	 * @param payload   消息内容
	 * @return
	 */
	public static boolean publish(String topicName, byte[] payload) {
		try {
			MqttTopic topic = client.getTopic(topicName);
			MqttMessage message = new MqttMessage();
			message.setQos(1);
			message.setRetained(false);
			message.setPayload(payload);
			MqttDeliveryToken token = topic.publish(message);
			System.out.println("mqtt消息发送结果等待。。。。。");
			token.waitForCompletion();
			System.out.println("mqtt消息发送结果：" + token.isComplete());
			return token.isComplete();
		} catch (Exception e) {
			e.printStackTrace();
			System.out.println("mqtt消息发送"+ e);
			return false;
		}
	}
}
